package com.atguigu.flink.sql.catalog;

import org.apache.flink.connector.jdbc.catalog.MySqlCatalog;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

import java.util.Arrays;

/**
 * Created by Smexy on 2023/4/14
 *
 *
 *  JDBCCatalog ： 通过读取JDBC的数据库，获取表的元数据信息。
 *          目前仅仅支持 Postgresql 及 Mysql
 *
 *
 *       作用:
 *              MySqlCatalog 连接上Mysql的某个库，读取库下已经存在的表！
 *
 *              一个Catalog必须实现 Catalog类的creaetTables()方法，才能把建表语句存储到外部设备。
 *              MySqlCatalog 没有实现，所以，MySqlCatalog不能保存你自己定义的表！
 *
 *
 */
public class Demo2_JDBCCatalog
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();

        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        //默认还是 GenericInMemoryCatalog，创建
        MySqlCatalog mySqlCatalog = new MySqlCatalog(
            Demo2_JDBCCatalog.class.getClassLoader(),
            "mysql",
            //表存放的库名
            "221109",
            "root",
            "000000",
            //baseUrl ，除了库名的 url
            "jdbc:mysql://hadoop102:3306"
        );

        //切换Catalog，
        tableEnvironment.registerCatalog("aa",mySqlCatalog);
        //注册的名字
        tableEnvironment.useCatalog("aa");

        //aa
        System.out.println(tableEnvironment.getCurrentCatalog());
        //221109
        System.out.println(tableEnvironment.getCurrentDatabase());

        //打印数组最优雅的方式
        System.out.println(Arrays.toString(tableEnvironment.listTables()));


       String createTableSql = " create table t1 ( id STRING, ts BIGINT , vc INT ," +
            "  `file.path` STRING NOT NULL METADATA " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/sensor.json'," +
            "  'format' = 'json'" +
            ") ";

       // Could not execute CreateTable in path `aa`.`221109`.`t1`
        tableEnvironment.executeSql(createTableSql);

        tableEnvironment.sqlQuery(" select * from emp   ").execute().print();

    }
}
